package com.example.rabbitmqboot.config;

import com.example.rabbitmqboot.converter.FileMessageConverter;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
// 对RabbitMQ连接池开启不可路由消息方法
// 关闭@RabbitListener默认开启的ACK响应
// 连接池配置
public class RabbitMQConfig {

    @Autowired
    private RabbitProperties rabbitProperties;


    @Bean
    //RabbitMQ连接池，从配置文件读取参数
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();

        cachingConnectionFactory.setHost(rabbitProperties.getHost());
        cachingConnectionFactory.setPort(rabbitProperties.getPort());
        cachingConnectionFactory.setUsername(rabbitProperties.getUsername());
        cachingConnectionFactory.setPassword(rabbitProperties.getPassword());
        cachingConnectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
        cachingConnectionFactory.setPublisherReturns(rabbitProperties.isPublisherReturns()); // 回传
        cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);//发布确认类型 开启回调机制

        return cachingConnectionFactory;
    }

    @Bean
    // RabbitListener使用连接池，listener.simple里面的不生效，这里设置一下
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();
        simpleRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);//手动确认
        simpleRabbitListenerContainerFactory.setConnectionFactory(connectionFactory);//使用上面配置的工厂
        simpleRabbitListenerContainerFactory.setPrefetchCount(1); //限流
        simpleRabbitListenerContainerFactory.setMaxConcurrentConsumers(5); //每个监听器的最大并发数量
        simpleRabbitListenerContainerFactory.setConcurrentConsumers(1); //每个监听器默认处理信息的线程数
        simpleRabbitListenerContainerFactory.setMessageConverter(new Jackson2JsonMessageConverter()); //json数据反序列化
        return simpleRabbitListenerContainerFactory;
    }


    private static final String EXCHANGE_NAME = "StudentExchange";
    private static final String QUEUE_NAME = "StudentQueue";
    private static final String ROUTING_KEY = "Student.#";


    @Bean
    public TopicExchange studentExchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    public Queue studentQueue() {
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    @Bean
    public Binding studentBind(@Qualifier("studentExchange") TopicExchange topicExchange,
                               @Qualifier("studentQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(topicExchange).with(ROUTING_KEY);
    }


    private static final String FILE_QUEUE_NAME = "FileQueue";
    private static final String FILE_ROUTING_KEY = "File.#";

    @Bean
    public Queue fileQueue() {
        return QueueBuilder.durable(FILE_QUEUE_NAME).build();
    }

    @Bean
    public Binding FileBind(@Qualifier("studentExchange") TopicExchange topicExchange,
                            @Qualifier("fileQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(topicExchange).with(FILE_ROUTING_KEY);
    }

//    @Bean
//    //注入自定义消息转换器
//    public FileMessageConverter fileMessageConverter() {
//        return new FileMessageConverter();
//    }


}
